package com.mars.ms.threadpool;

import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * author by: mars
 * Date: 2022/3/21 9:15
 * Description:  自定义可暂停的线程池, 钩子方法，前后做一些事情
 * 1.继承 ThreadPoolExecutor
 * 2.
 */
public class PauseThreadPoolDemo extends ThreadPoolExecutor {

    private boolean isPaused;
    private final ReentrantLock reentrantLock = new ReentrantLock();
    private final Condition condition = reentrantLock.newCondition();

    public PauseThreadPoolDemo(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public PauseThreadPoolDemo(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public PauseThreadPoolDemo(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public PauseThreadPoolDemo(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    // 执行前做一些事情
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        reentrantLock.lock();
        try {
            while (isPaused) {
                condition.await();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            reentrantLock.unlock();
        }
    }

    // 执行后做一些事情
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
    }

    // 暂停
    private void pause(){
        reentrantLock.lock();
        try{
            isPaused = true;
        } finally {
            reentrantLock.unlock();
        }
    }

    // 恢复函数
    private void resume() {
        reentrantLock.lock();
        try {
            isPaused = false;
            // 唤醒
            condition.signalAll();
        } finally {
            reentrantLock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        PauseThreadPoolDemo pausePool = new PauseThreadPoolDemo(10,20,10l,TimeUnit.SECONDS,new LinkedBlockingDeque<>());
        Runnable runTask = new Runnable(){
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + " 我被执行了");
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        for (int i = 0; i < 100000; i++) {
            pausePool.execute(runTask);
        }

        // 睡眠一段时间 暂停
        Thread.sleep(1500);
        pausePool.pause();
        System.out.println("线程池被暂停了");
        Thread.sleep(1000);
        System.out.println("线程池被恢复了");
        pausePool.resume();
    }
}
